package com.wsjj.gmall.dim.function;

import com.alibaba.fastjson.JSONObject;
import com.wsjj.gmall.bean.TableProcessDim;
import com.wsjj.gmall.constant.Constant;
import com.wsjj.gmall.util.HbaseUtil;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.client.Connection;

/**
 * 该类的作用是：将维度数据写入到hbase中
 */
public class HBaseSinkFunction extends RichSinkFunction<Tuple2<JSONObject, TableProcessDim>> {
    Connection hBseConnection = null;
    @Override
    public void open(Configuration parameters) throws Exception {
        hBseConnection = HbaseUtil.getHBseConnection();

    }

    //            写入数据到hbase中
    @Override
    public void invoke(Tuple2<JSONObject, TableProcessDim> value) throws Exception {
        JSONObject f0 = value.f0;
        TableProcessDim f1 = value.f1;


//                插入数据
        Object type = f0.get("type");
        f0.remove("type");


        if("delete".equals(type)){


            HbaseUtil.delRow(hBseConnection, Constant.HBASE_NAMESPACE,f1.getSinkTable(),f0.getString(f1.getSinkRowKey()));

        }else {
            HbaseUtil.putRow(hBseConnection, Constant.HBASE_NAMESPACE,f1.getSinkTable(),f0.getString(f1.getSinkRowKey()),f1.getSinkFamily(),f0);
        }



//                SinkFunction.super.invoke(value);
    }


    @Override
    public void close() throws Exception {
        HbaseUtil.closeHBaseConn(hBseConnection);
    }
}
